package com.neusoft.bigdata.hbase.biz.impl;

import java.io.IOException;
import java.util.Calendar;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;

import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;

import com.neusoft.bigdata.hbase.biz.AbstractHBaseBiz;
import com.neusoft.bigdata.hbase.biz.MonitorService;
import com.neusoft.bigdata.hbase.model.MetricData;

/**
 * 从HBase中，Metric表中直接查询监控数据接口
 * 
 * @author lihzh
 * @date 2012-11-21 下午6:28:23
 */
public class MonitorServiceHBaseImpl extends AbstractHBaseBiz implements
		MonitorService {

	private static HTable metricTable;

	static {
		try {
			metricTable = new HTable(conf, TABLE_NAME_METRIC);
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	@Override
	public Map<String, MetricData>  getDataByFQNAndTimeRange(String fqn,
			long startTime, long endTime) {
		Calendar startCalendar = Calendar.getInstance();
		startCalendar.setTimeInMillis(startTime);
		Calendar endCalendar = Calendar.getInstance();
		endCalendar.setTimeInMillis(endTime);
		int dayCount = endCalendar.get(Calendar.DAY_OF_YEAR) - startCalendar.get(Calendar.DAY_OF_YEAR);
		int startHourOfDay = startCalendar.get(Calendar.HOUR_OF_DAY);
		int endHourOfDay = endCalendar.get(Calendar.HOUR_OF_DAY);
		int pointCount = endCalendar.get(Calendar.HOUR_OF_DAY);
		if (dayCount != 0) {
			pointCount = (dayCount - 1) * 24 + 24 - startHourOfDay + endHourOfDay;
		} else {
			pointCount = endHourOfDay - startHourOfDay;
		}
		try {
			return calSingle(fqn, startTime, endTime, pointCount);
		} catch (IOException e) {
			e.printStackTrace();
		}
		return null;
	}

	private Map<String, MetricData>  calSingle(String fqn, long startTime,
			long endTime, int pointCount) throws IOException {
		Map<String, MetricData> metricMap = new HashMap<String, MetricData>();
		Calendar tempCalendar = Calendar.getInstance();
		long tempStart = startTime; 
		long tempEnd = 0;
		MetricData cpuloadData = new MetricData(COLUMN_NAME_CPULOAD);
		MetricData memoryUsageData = new MetricData(COLUMN_NAME_MEMORYUSAGE);
		for (int i = 0; i < pointCount - 1; i ++) {
			tempCalendar.setTimeInMillis(tempStart);
			if (tempCalendar.get(Calendar.HOUR_OF_DAY) < 23) {
				tempCalendar.add(Calendar.HOUR_OF_DAY, 1);
			} else {
				tempCalendar.add(Calendar.DAY_OF_YEAR, 1);
				tempCalendar.set(Calendar.HOUR_OF_DAY, 0);
			}
			tempEnd = tempCalendar.getTimeInMillis();
			String startRowKey = creatRowKey(fqn, tempStart);
			String endRowKey = creatRowKey(fqn, tempEnd);
			Iterator<Result> resultIt = getByFQN(metricTable, startRowKey, endRowKey);
			Map<String, Double> avgMap = calculResultAvg(resultIt);
			cpuloadData.addDataPoint(avgMap.get(COLUMN_NAME_CPULOAD), tempStart);
			memoryUsageData.addDataPoint(avgMap.get(COLUMN_NAME_MEMORYUSAGE), tempStart);
			tempStart = tempEnd;
		}
		metricMap.put(COLUMN_NAME_CPULOAD, cpuloadData);
		metricMap.put(COLUMN_NAME_MEMORYUSAGE, memoryUsageData);
		return metricMap;
	}

	private String creatRowKey(String fqn, long time) {
		return fqn + "_" + time;
	}

	/**
	 * 计算各指标平均值<br>
	 * TODO 扩展更多的指标计算
	 * 
	 * @param resultIt
	 * @author OneCoder(OneCoder)
	 * @date 2012-11-19 下午11:15:32
	 */
	private Map<String, Double> calculResultAvg(Iterator<Result> resultIt) {
		Map<String, Double> avgMap = new HashMap<String, Double>();
		int cpuloadCount = 0;
		double cpuloadSum = 0;
		double cpuloadAvg = 0d;
		int memoryUsageCount = 0;
		double memoryUsageSum = 0d;
		double memoryUsageAvg = 0d;
		while (resultIt.hasNext()) {
			Result result = resultIt.next();
			KeyValue cpuloadKv = result.getColumnLatest(
					Bytes.toBytes(FAMILY_NAME_METRICS),
					Bytes.toBytes(COLUMN_NAME_CPULOAD));
			if (cpuloadKv != null && !cpuloadKv.isEmptyColumn()) {
				Double cpuloadValue = Double.valueOf(new String(cpuloadKv
						.getValue()));
				cpuloadCount++;
				cpuloadSum += cpuloadValue;
			}
			KeyValue memoryUsageKv = result.getColumnLatest(
					Bytes.toBytes(FAMILY_NAME_METRICS),
					Bytes.toBytes(COLUMN_NAME_MEMORYUSAGE));
			if (memoryUsageKv != null && !memoryUsageKv.isEmptyColumn()) {
				Double memoryUsageValue = Double.valueOf(new String(
						memoryUsageKv.getValue()));
				memoryUsageCount++;
				memoryUsageSum += memoryUsageValue;
			}
		}
		cpuloadAvg = cpuloadSum / cpuloadCount;
		memoryUsageAvg = memoryUsageSum / memoryUsageCount;
		avgMap.put(COLUMN_NAME_CPULOAD, cpuloadAvg);
		avgMap.put(COLUMN_NAME_MEMORYUSAGE, memoryUsageAvg);
		return avgMap;
	}

}
